package com.fsxgt.datax;

import org.apache.commons.lang3.Validate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;

import com.alibaba.datax.common.plugin.AbstractTaskPlugin;
import com.alibaba.datax.common.plugin.RecordReceiver;
import com.alibaba.datax.common.spi.Writer;
import com.alibaba.datax.core.taskgroup.runner.WriterRunner;

/**
 * Created by jingxing on 14-9-1.
 * <p/>
 * 单个slice的writer执行调用
 */
public class FastWriterRunner extends WriterRunner implements Runnable {

    private static final Logger LOG = LoggerFactory
            .getLogger(FastWriterRunner.class);

    private RecordReceiver recordReceiver;
    private String logFileName;

    public void setRecordReceiver(RecordReceiver receiver) {
        this.recordReceiver = receiver;
    }

    public FastWriterRunner(AbstractTaskPlugin abstractTaskPlugin) {
        super(abstractTaskPlugin);
        if(MDC.get("logFileName")!=null) {
    		logFileName=MDC.get("logFileName");
    	}
    }

    @Override
    public void run() {
    	
    	if(logFileName!=null) {
    		MDC.put("logFileName", logFileName);
    	}
    	
        Validate.isTrue(this.recordReceiver != null);

        Writer.Task taskWriter = (Writer.Task) this.getPlugin();
        //统计waitReadTime，并且在finally end
        try {
            LOG.debug("task writer starts to do init ...");
            taskWriter.init();

            

//            LOG.debug("task writer starts to do prepare ...");
//            taskWriter.prepare();

            LOG.debug("task writer starts to write ...");
            taskWriter.startWrite(recordReceiver);

//            LOG.debug("task writer starts to do post ...");
//            taskWriter.post();

            super.markSuccess();
        } catch (Throwable e) {
            LOG.error("Writer Runner Received Exceptions:", e);
            recordReceiver.shutdown();
            super.markFail(e);
        } finally {
            LOG.debug("task writer starts to do destroy ...");
            super.destroy();
        }
    }
    
    public boolean supportFailOver(){
    	Writer.Task taskWriter = (Writer.Task) this.getPlugin();
    	return taskWriter.supportFailOver();
    }

    public void shutdown(){
        recordReceiver.shutdown();
    }
}
